package beat

import (
	"encoding/json"
	"logbeat/internal/ectx"
	"logbeat/internal/req"
	"path/filepath"
	"strings"
	"time"

	"github.com/hpcloud/tail"
	"go.uber.org/zap"
)

type Beat struct {
	ectx    *ectx.Ectx
	config  tail.Config
	backtxt string
	txtchan chan [2]string
}

func NewBeat(ctx *ectx.Ectx) *Beat {
	a := &Beat{ectx: ctx, config: tail.Config{
		ReOpen:    true,
		Follow:    true,
		Location:  &tail.SeekInfo{Offset: 0, Whence: 2},
		MustExist: false,
		Poll:      true,
	}, txtchan: make(chan [2]string)}
	go a.req()
	return a
}

func (this *Beat) GetFiles(path string) []string {
	m, err := filepath.Glob(path)
	if err != nil {
		panic(err)
	}
	return m
}

func (this *Beat) Tail() {
	this.ectx.Log.Debug("tail start", zap.Any("paths", this.ectx.Config.Beat.Files))
	for _, v := range this.ectx.Config.Beat.Files {
		this.ectx.Log.Debug("File path:" + v.File)
		files := this.GetFiles(v.File)
		for _, file := range files {
			this.ectx.Log.Debug("start listen File:" + file)
			go this.listen(v.Index, v.File)
		}
	}
}

func (this *Beat) listen(k string, f string) {
	t, err := tail.TailFile(f, this.config)

	if err != nil {
		this.ectx.Log.Error("启动失败:", zap.String("index", k), zap.String("file", f), zap.Error(err))
	} else {
		this.ectx.Log.Info("监听文件", zap.String("index", k), zap.String("file", f))
	}
	for {
		select {
		case txt := <-t.Lines:
			this.txtchan <- [2]string{k, txt.Text}
		}
	}
}

func (this *Beat) req() {
	for {
		txt := <-this.txtchan
		url := this.ectx.Config.Beat.Hosts + "/api/" + this.GetIndex(txt[0]) + "/_doc"
		if err := this.retry(url, txt); err != nil {
			this.ectx.Log.Error("error", zap.Error(err), zap.String("index", txt[0]), zap.String("url", url))
			this.loop(url, txt)
		}
	}
}

func (this *Beat) retry(url string, txt [2]string) error {
	var (
		err error
		num int
	)
	for {
		if num == 3 {
			return err
		}
		bd, status, _err := req.Req(this.ectx, "POST", url, strings.NewReader(this.GetDoc(txt)))
		if _err != nil {
			num += 1
			err = _err
			continue
		}
		if status != 200 {
			this.ectx.Log.Error("response_not_200", zap.String("txt", this.GetDoc(txt)), zap.String("body", bd), zap.Int("status", status))
		}
		return nil
	}
}

func (this *Beat) loop(url string, txt [2]string) {
	for {
		select {
		case <-time.After(time.Duration(this.ectx.Config.Beat.RetrySecond) * time.Second):
			bd, status, err := req.Req(this.ectx, "POST", url, strings.NewReader(this.GetDoc(txt)))
			if err == nil {
				if status != 200 {
					this.ectx.Log.Error("response_not_200", zap.String("body", bd), zap.Int("status", status))
				}
				return
			}
		}
	}
}

func (this *Beat) GetIndex(index string) string {
	return index + "_" + time.Now().Month().String()
}

func (this *Beat) GetDoc(txt [2]string) string {

	startindex := strings.Index(txt[1], "{")
	if startindex != -1 {
		endindex := strings.Index(txt[1], "}")
		if endindex != -1 {
			return txt[1][startindex : endindex+1]
		}
	}

	a, _ := json.Marshal(map[string]string{
		"doc": txt[1],
	})

	return string(a)
}
